#include "queue.h"

namespace ai_service
{

Ret AsyncTaskBase::CommitTo(AsyncTaskQueue& queue,
    AsyncTaskParams& params)
{
#ifdef TASK_QUEUE_WITH_MAX_LENGTH
    constexpr int taskQueueMaxSize = 20;    
    if (queue.size() >= taskQueueMaxSize) {
        return RET_TASK_QUEUE_FULL;
    }
#endif  // TASK_QUEUE_WITH_MAX_LENGTH

    TaskExec task = std::bind(&AsyncTaskBase::Execute, this, params);
    queue.Push(std::make_pair<AsyncTaskBase*, TaskExec>(this, std::move(task)));
    this->m_perf.update();
    return Ret::RET_OK;
}

void AsyncTaskBase::PreCheck(AsyncTaskParams& params)
{
    // 任务执行前
    // 判断输入数据是否准备好
    for (auto& param: params.input) {
        param->WaitStatus(STATUS_READY);
        param->UpdateStatus(m_name);
    }

    // 判断输出数据是否已经FREE
    for (auto& param: params.output) {
        param->WaitStatus(STATUS_FREE);
        param->UpdateStatus(m_name);
    }
}

void AsyncTaskBase::PostUpdate(AsyncTaskParams& params)
{
    // 执行完成之后
    // 更新输入状态为Free
    for (auto& param: params.input) {
        param->UpdateStatus(m_name);
    }
    // 更新输出状态为Ready
    for (auto& param: params.output) {
        param->UpdateStatus(m_name);
    }
}

Ret AsyncTaskBase::ExecuteTask(AsyncTaskParams& params)
{
    return Ret::RET_OK;
}

Ret AsyncTaskBase::Execute(AsyncTaskParams& params)
{
    Ret ret = Ret::RET_OK;

    // 任务执行前
    PreCheck(params);

    // 任务执行
    LOG_INFO("Task: %s Execute", this->GetTaskName().c_str());
    ret = ExecuteTask(params);
    if (ret != Ret::RET_OK) {
        LOG_ERROR("Task: %s Execute failed: ret: %d", this->GetTaskName().c_str(), ret);
    }
    // 任务执行后
    PostUpdate(params);

    return ret;
}

Ret AsyncTaskBase::Sync()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_cond.wait(lock);
    return Ret::RET_OK;
}

Ret AsyncTaskBase::Notify()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_cond.notify_one();
    return Ret::RET_OK;
}

Ret AsyncTaskBase::Init(const Config& config)
{
    return Ret::RET_OK;
}

Ret AsyncTaskBase::Destroy()
{
    return Ret::RET_OK;
}

Ret AsyncTaskQueue::Push(std::pair<AsyncTaskBase*, TaskExec> task)
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_queue.push(task);
    m_cond.notify_one();  // Notify the loop thread
    return Ret::RET_OK;
}

Ret AsyncTaskQueue::Pop(std::pair<AsyncTaskBase*, TaskExec>& task)
{
    std::unique_lock<std::mutex> lock(m_mutex);
    if (m_queue.empty()) {
        return RET_TASK_QUEUE_EMPTY;
    }
    task = m_queue.front();
    m_queue.pop();
    return Ret::RET_OK;
}

Ret AsyncTaskQueue::Wait()
{
    std::unique_lock<std::mutex> lock(m_mutex);
    m_cond.wait(lock);
    return Ret::RET_OK;
}


Ret AsyncTaskQueue::Loop(bool& flag)
{
    std::thread thread([&] {
        Ret ret = RET_OK;
        pthread_setname_np(pthread_self(), m_name.c_str());
        while (flag) {
            std::pair<AsyncTaskBase*, TaskExec> op;
            while (Pop(op) == RET_TASK_QUEUE_EMPTY) {
                Perf timer;
                Wait();
                timer.tick("AsyncTaskQueue Wait");
            }
            {
                auto cls = op.first;
                cls->ElapseFromCommit2Exec();
            
                Perf timer;
                ret = op.second();
                timer.tick(cls->GetTaskName());
                if (ret != RET_OK) {
                    LOG_ERROR("Task: %s exec failed: ret: %d", cls->GetTaskName().c_str(), ret);
                } else {
                    cls->Notify();
                    LOG_INFO("Task: %s exec success", cls->GetTaskName().c_str());
                }
            }
        }
        LOG_WARN("AsyncTaskQueue: %s loop exit", m_name.c_str());
    });
    thread.detach();
    return Ret::RET_OK;
}


}